from kafka import  KafkaConsumer
import time
import json

import sumo_main
from constants import KAFKA_HOST, VIR_DEVICE_STATUS_TOPIC
def receive_message(userId, task_id):
    print('kafka 消费被触发')
    # 首先定义全局变量，用于在不同方法内互相通信
    global vir_gantry_dict, vir_ramp_light, SPEED_LIMIT_SIGN, \
        speed_limit_point_list, speed_limit_list, speed_gantry_asc, \
        emergency_dict, special_dict
    key = str(task_id)

    # key = str(579) + '_' + str(224) + '_' + str(1)
    consumer = KafkaConsumer(
        VIR_DEVICE_STATUS_TOPIC,
        bootstrap_servers=KAFKA_HOST,
        auto_offset_reset='latest',
        enable_auto_commit=False
    )
    # 将消费者的偏移量设置为最后一条消息的偏移量
    """
            同步发送 数据
            :param data_li:  发送数据
            :return:
            """
    while True:
        time.sleep(1)
        for message in consumer:
            if message:
                message_key = message.key.decode('utf-8')
                message_value = message.value.decode('utf-8')
                vir_gantry_light = json.loads(message_value)
                print("kafka消费者接收到的数据", message_key, message_value)
                if 'taskId' in vir_gantry_light:
                    if vir_gantry_light['taskId'] == key:
                        if 'frameStrategies' in vir_gantry_light:
                            vir_gantry_frame = vir_gantry_light['frameStrategies']
                            for vir_gantry in vir_gantry_frame:
                                gantryId = vir_gantry['frameId']
                                speedLimitValue = vir_gantry['speedLimitValue']
                                emergencyLight = vir_gantry['emergencyLight']
                                specialLight = vir_gantry['specialLight']
                                emergency_json = {
                                    "emergencyLight": int(emergencyLight)
                                }
                                special_json = {
                                    "specialLight": int(specialLight)
                                }
                                sumo_main.SPEED_LIMIT_SIGN = True
                                for i in range(len(sumo_main.speed_gantry_asc)):
                                    if gantryId in sumo_main.speed_gantry_asc[i]:
                                        index = sumo_main.speed_gantry_asc[i].index(gantryId)
                                        sumo_main.speed_limit_list[i][index] = int(speedLimitValue)
                                sumo_main.emergency_dict[gantryId] = emergency_json
                                sumo_main.special_dict[gantryId] = special_json
                                # sumo_main.emergency_dict=emergency_dict
                                # sumo_main.special_dict=special_dict
                    if 'rampStrategy' in vir_gantry_light:
                        rampStrategy = vir_gantry_light['rampStrategy']
                        if 'rampLight' in rampStrategy:
                            vir_ramp_light = int(rampStrategy['rampLight'])
                            sumo_main.vir_ramp_light=vir_ramp_light